package net.bwie.realtime.jtp.Rpfunction;

import org.apache.commons.lang3.time.FastDateFormat;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.List;

/**
 * 简介说明:
 *
 * @author: LiLi
 * @date: 2025/05/28 10:41:29
 * @version: 1.0
 */
public class TopNHotStateFunction extends KeyedProcessFunction<Long, Tuple4<Long, Long, Integer, Long>, String> {
    private final int topSize;
    private transient ValueState<List<Tuple4<Long, Long, Integer, Long>>> state;
    private FastDateFormat format = FastDateFormat.getInstance("yyyy-MM-dd HH:mm:ss");
    public TopNHotStateFunction(int topSize) {
        this.topSize = topSize;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        ValueStateDescriptor<List<Tuple4<Long, Long, Integer, Long>>> descriptor =
                new ValueStateDescriptor<>(
                        "top-items-state",
                        TypeInformation.of(new TypeHint<List<Tuple4<Long, Long, Integer, Long>>>() {})
                );
        state = getRuntimeContext().getState(descriptor);
    }

    @Override
    public void processElement(Tuple4<Long, Long, Integer, Long> value,
                               Context ctx,
                               Collector<String> out) throws Exception {

        // 获取当前状态
        List<Tuple4<Long, Long, Integer, Long>> current = state.value();
        if (current == null) {
            current = new ArrayList<>();
        }

        // 添加新记录
        current.add(value);

        // 按销量降序排序
        current.sort((o1, o2) -> o2.f3.compareTo(o1.f3));

        // 保留TopN
        if (current.size() > topSize) {
            current = current.subList(0, topSize);
        }

        // 更新状态
        state.update(current);

        // 注册定时器
        ctx.timerService().registerEventTimeTimer(value.f1 + 1);

        String windowStartTime = format.format(value.f0);
        String windowEndTime = format.format(value.f1);

        String output = windowStartTime + "," + windowEndTime + "," + value.f2 + "," +
                value.f3 + "," + System.currentTimeMillis();

        out.collect(output);
    }
}
